package com.sheepone.distributor.thread;

import java.util.concurrent.*;

/**
 * @author Miss.杨
 */
public class LoadExecutorHolder {

    private final Semaphore semaphore;

    private final ThreadPoolExecutor threadPoolExecutor;

    public LoadExecutorHolder(String name, int permits, int keepAliveTime) {
        // 默认 loadFactor = 1 时可保证线程安全（不出现不安全现象）
        this(name, permits, keepAliveTime, 1);
    }

    public LoadExecutorHolder(String name, int permits, int keepAliveTime, double loadFactor) {
        semaphore = new Semaphore(permits);
        threadPoolExecutor = new ThreadPoolExecutor(
                permits,
                (int) (permits * (1 + loadFactor)),
                keepAliveTime,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(),
                NamedDaemonThreadFactory.getInstance(name)
        );
    }

    public int getActiveCount() {
        return threadPoolExecutor.getActiveCount();
    }

    public void submit(Runnable task) {
        // semaphore本身并不会将信号量与当前线程绑定。
        // 它只是简单地从信号量中获取一个许可，并且不会因为线程中断而停止尝试获取许可
        semaphore.acquireUninterruptibly();
        threadPoolExecutor.submit(() -> {
            try {
                task.run();
            } finally {
                semaphore.release();
                // 不安全现象：
                // 可能存在 permits 个线程在这里同时释放 semaphore，同时另外还有 >= permits 个线程 在 acquire 中
                // 这将导致 原来持有 semaphore 的线程还未结束，而 等待的线程获取到 semaphore 后 涌入（最多涌入 permits 个）
                // 导致 ThreadPool 满载，新来的线程被拒绝从而抛出异常
            }
        });
    }

    public void shotDown() {
        threadPoolExecutor.shutdown();
    }
}